package com.shujia.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.client.{HConnection, HConnectionManager, HTableInterface}


class HbaseSink[IN](tableName: String, mapper: HbaseMapper[IN]) extends RichSinkFunction[IN] {


  val _mapper: HbaseMapper[IN] = mapper

  val _tableName: String = tableName

  var connection: HConnection = _

  var tableInterface: HTableInterface = _

  override def open(parameters: Configuration): Unit = {
    //换件hbase连接
    val configuration = new org.apache.hadoop.conf.Configuration()
    configuration.set("hbase.zookeeper.quorum", "node2:2181,node3:2181,node4:2181")
    connection = HConnectionManager.createConnection(configuration)
    tableInterface = connection.getTable(_tableName)
  }

  override def close(): Unit = {

    if (connection != null) {
      connection.close()
    }
  }

  override def invoke(kv: IN, context: SinkFunction.Context[_]): Unit = {
    val put = mapper.getPut(kv)
    tableInterface.put(put)
  }
}
